package org.example.demo.broker;

import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import org.example.demo.common.convert.GenericMessageConverter;
import org.example.demo.common.convert.RabbitMessageConverter;
import org.example.demo.common.serializer.Serializer;
import org.example.demo.common.serializer.SerializerFactory;
import org.example.demo.common.serializer.impl.SerializerFactoryImpl;
import org.example.demo.common.serializer.SerializerVariety;
import org.example.demo.domain.Message;
import org.example.demo.domain.MessageType;
import org.example.demo.domain.exception.MessageRunTimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitTemplateContainer.class);

    private final Map<String /* TOPIC **/, RabbitTemplate> rabbitTemplateMap = Maps.newConcurrentMap();
    private static final Splitter HASH_SPLITTER = Splitter.on("#");
    private static final SerializerFactory SERIALIZER_FACTORY = SerializerFactoryImpl.getInstance();

    @Autowired
    private ConnectionFactory connectionFactory;

    public RabbitTemplate getTemplate(Message message) throws MessageRunTimeException {
        if (message == null) {
            throw new MessageRunTimeException("message is null");
        }
        String exchange = message.getExchange();
        RabbitTemplate template = this.rabbitTemplateMap.get(exchange);
        if (template != null) {
            return template;
        }
        LOGGER.info("RabbitTemplateContainer#getTemplate exchange {} not exists, create one", exchange);
        RabbitTemplate newTemplate = new RabbitTemplate(this.connectionFactory);
        newTemplate.setExchange(message.getExchange());
        newTemplate.setRoutingKey(message.getRoutingKey());
        newTemplate.setRetryTemplate(new RetryTemplate());
        Serializer serializer = SERIALIZER_FACTORY.create(SerializerVariety.JSON, Message.class);
        GenericMessageConverter gmc = new GenericMessageConverter(serializer);
        RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
        newTemplate.setMessageConverter(rmc);
        String msgType = message.getMessageType();
        if (!msgType.equals(MessageType.RAPID)) {
            newTemplate.setConfirmCallback(this);
        }
        this.rabbitTemplateMap.putIfAbsent(exchange, newTemplate);
        return this.rabbitTemplateMap.get(exchange);
    }


    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        Preconditions.checkNotNull(correlationData);
        List<String> info = HASH_SPLITTER.splitToList(correlationData.getId());
        String messageId = info.get(0);
        Long sendTime = Long.parseLong(info.get(1));
        String msgType = info.get(2);
        if (ack) {
            LOGGER.info("send message is ok, confirm messageId: {}, sendTime: {}", messageId, sendTime);
        } else {
            LOGGER.error("send message failed, confirm messageId: {}, sendTime: {}, cause: {}", messageId, sendTime, cause);
        }
    }
}
